import pandas as pd
import numpy as np
from datetime import datetime

def clean_data(df):
    """
    数据清洗函数
    
    参数:
    - df: 待清洗的DataFrame
    
    返回:
    - 清洗后的DataFrame
    """
    # 复制数据以避免修改原始数据
    cleaned_df = df.copy()
    
    # 处理缺失值
    for col in cleaned_df.columns:
        if cleaned_df[col].dtype == 'object':
            # 对于字符串列，用众数填充
            cleaned_df[col].fillna(cleaned_df[col].mode()[0], inplace=True)
        else:
            # 对于数值列，用均值填充
            cleaned_df[col].fillna(cleaned_df[col].mean(), inplace=True)
    
    # 处理日期列
    date_columns = [col for col in cleaned_df.columns if 'date' in col.lower()]
    for col in date_columns:
        cleaned_df[col] = pd.to_datetime(cleaned_df[col], errors='coerce')
        # 对于无法转换的日期，使用NaN填充，然后用前后值插值
        cleaned_df[col].interpolate(method='time', inplace=True)
    
    # 处理异常值（使用IQR方法）
    numeric_columns = cleaned_df.select_dtypes(include=[np.number]).columns
    for col in numeric_columns:
        if col not in date_columns:  # 跳过日期列
            q1 = cleaned_df[col].quantile(0.25)
            q3 = cleaned_df[col].quantile(0.75)
            iqr = q3 - q1
            lower_bound = q1 - 1.5 * iqr
            upper_bound = q3 + 1.5 * iqr
            
            # 用边界值替换异常值
            cleaned_df[col] = np.where(cleaned_df[col] < lower_bound, lower_bound, cleaned_df[col])
            cleaned_df[col] = np.where(cleaned_df[col] > upper_bound, upper_bound, cleaned_df[col])
    
    return cleaned_df

def transform_data(df, transformations):
    """
    数据转换函数
    
    参数:
    - df: 待转换的DataFrame
    - transformations: 转换规则列表
    
    返回:
    - 转换后的DataFrame
    """
    transformed_df = df.copy()
    
    for transform in transformations:
        action = transform.get('action')
        column = transform.get('column')
        
        if not action or not column or column not in transformed_df.columns:
            continue
        
        if action == 'normalize':
            # 归一化处理
            min_val = transformed_df[column].min()
            max_val = transformed_df[column].max()
            if max_val > min_val:
                transformed_df[column] = (transformed_df[column] - min_val) / (max_val - min_val)
        
        elif action == 'standardize':
            # 标准化处理
            mean_val = transformed_df[column].mean()
            std_val = transformed_df[column].std()
            if std_val > 0:
                transformed_df[column] = (transformed_df[column] - mean_val) / std_val
        
        elif action == 'log_transform':
            # 对数转换（处理正值）
            if transformed_df[column].min() > 0:
                transformed_df[column] = np.log(transformed_df[column])
        
        elif action == 'bin':
            # 分箱处理
            bins = transform.get('bins', 5)
            labels = transform.get('labels')
            transformed_df[f'{column}_bin'] = pd.cut(
                transformed_df[column], 
                bins=bins, 
                labels=labels
            )
        
        elif action == 'create_ratio':
            # 创建比率列
            numerator = transform.get('numerator')
            denominator = transform.get('denominator')
            if numerator and denominator and numerator in transformed_df.columns and denominator in transformed_df.columns:
                new_col = f'{numerator}_over_{denominator}'
                transformed_df[new_col] = transformed_df[numerator] / transformed_df[denominator].replace(0, 1e-10)
        
        elif action == 'extract_date_part':
            # 提取日期部分
            part = transform.get('part', 'year')
            if transformed_df[column].dtype == 'datetime64[ns]':
                new_col = f'{column}_{part}'
                if part == 'year':
                    transformed_df[new_col] = transformed_df[column].dt.year
                elif part == 'month':
                    transformed_df[new_col] = transformed_df[column].dt.month
                elif part == 'day':
                    transformed_df[new_col] = transformed_df[column].dt.day
                elif part == 'quarter':
                    transformed_df[new_col] = transformed_df[column].dt.quarter
                elif part == 'weekday':
                    transformed_df[new_col] = transformed_df[column].dt.weekday
    
    return transformed_df

def aggregate_data(df, group_by, aggregates):
    """
    数据聚合函数
    
    参数:
    - df: 待聚合的DataFrame
    - group_by: 分组列
    - aggregates: 聚合规则
    
    返回:
    - 聚合后的DataFrame
    """
    # 检查分组列是否存在
    valid_group_by = [col for col in group_by if col in df.columns]
    if not valid_group_by:
        return df
    
    # 构建聚合字典
    agg_dict = {}
    for agg in aggregates:
        col = agg.get('column')
        func = agg.get('function', 'sum')
        
        if col and col in df.columns:
            if func == 'sum':
                agg_dict[col] = 'sum'
            elif func == 'mean':
                agg_dict[col] = 'mean'
            elif func == 'count':
                agg_dict[col] = 'count'
            elif func == 'max':
                agg_dict[col] = 'max'
            elif func == 'min':
                agg_dict[col] = 'min'
            elif func == 'median':
                agg_dict[col] = 'median'
    
    if not agg_dict:
        return df
    
    # 执行聚合
    aggregated_df = df.groupby(valid_group_by).agg(agg_dict).reset_index()
    
    # 重命名列以明确聚合方式
    aggregated_df.columns = [
        f'{col}_{func}' if col in agg_dict else col 
        for col, func in zip(aggregated_df.columns, [None] * len(valid_group_by) + list(agg_dict.values()))
    ]
    
    return aggregated_df

def filter_data(df, filters):
    """
    数据过滤函数
    
    参数:
    - df: 待过滤的DataFrame
    - filters: 过滤条件列表
    
    返回:
    - 过滤后的DataFrame
    """
    filtered_df = df.copy()
    
    for filter_condition in filters:
        column = filter_condition.get('column')
        operator = filter_condition.get('operator')
        value = filter_condition.get('value')
        
        if not column or not operator or value is None or column not in filtered_df.columns:
            continue
        
        # 尝试将值转换为与列相同的数据类型
        try:
            if filtered_df[column].dtype == 'int64':
                value = int(value)
            elif filtered_df[column].dtype == 'float64':
                value = float(value)
            elif filtered_df[column].dtype == 'datetime64[ns]':
                value = pd.to_datetime(value)
        except (ValueError, TypeError):
            continue
        
        # 应用过滤条件
        if operator == 'equals':
            filtered_df = filtered_df[filtered_df[column] == value]
        elif operator == 'not_equals':
            filtered_df = filtered_df[filtered_df[column] != value]
        elif operator == 'greater_than':
            filtered_df = filtered_df[filtered_df[column] > value]
        elif operator == 'less_than':
            filtered_df = filtered_df[filtered_df[column] < value]
        elif operator == 'greater_than_or_equal':
            filtered_df = filtered_df[filtered_df[column] >= value]
        elif operator == 'less_than_or_equal':
            filtered_df = filtered_df[filtered_df[column] <= value]
        elif operator == 'contains':
            if filtered_df[column].dtype == 'object':
                filtered_df = filtered_df[filtered_df[column].str.contains(str(value), case=False)]
        elif operator == 'not_contains':
            if filtered_df[column].dtype == 'object':
                filtered_df = filtered_df[~filtered_df[column].str.contains(str(value), case=False)]
        elif operator == 'in':
            if isinstance(value, list):
                filtered_df = filtered_df[filtered_df[column].isin(value)]
        elif operator == 'not_in':
            if isinstance(value, list):
                filtered_df = filtered_df[~filtered_df[column].isin(value)]
    
    return filtered_df
